package com.atguigu.edu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.PropertyNamingStrategy;
import com.alibaba.fastjson.serializer.SerializeConfig;
import com.atguigu.edu.realtime.app.func.DimAsyncFunction;
import com.atguigu.edu.realtime.bean.DwdLearnPlayBean;
import com.atguigu.edu.realtime.bean.DwsLearnChapterPlayWindowBean;
import com.atguigu.edu.realtime.util.DateFormatUtil;
import com.atguigu.edu.realtime.util.DorisUtil;
import com.atguigu.edu.realtime.util.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

/**
 * 各章节视频播放
 */
public class DwsPlayPageView {
    public static void main(String[] args) {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(4);
        //TODO 2.检查点相关的设置
        env.enableCheckpointing(5000L);
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(1, Time.seconds(1L)));
        //TODO 3.从kafka的主题中读取数据
        //3.1 声明消费的主题以及消费者组
        String topic = "dwd_play_page_view";
        String groupId = "dws_learn_chapter_play_window";
        //3.2 创建消费者对象
        KafkaSource<String> kafkaSource = MyKafkaUtil.getKafkaSource(topic, groupId);
        //3.3 消费数据  封装为流
        DataStreamSource<String> kafkaDs = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka_source");
        kafkaDs.print("^^^");
        //TODO 4.对流中数据类型进行转换    jsonStr->jsonObj->实体类对象
        SingleOutputStreamOperator<DwsLearnChapterPlayWindowBean> dwdPlayBeanDs = kafkaDs.process(new ProcessFunction<String, DwsLearnChapterPlayWindowBean>() {
            @Override
            public void processElement(String value, ProcessFunction<String, DwsLearnChapterPlayWindowBean>.Context ctx, Collector<DwsLearnChapterPlayWindowBean> out) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                //DwdLearnPlayBean dwdPlayBean = jsonObject.getObject("DwdPlayBean", DwdLearnPlayBean.class);
                DwsLearnChapterPlayWindowBean bean = DwsLearnChapterPlayWindowBean.builder()
                        .userId(jsonObject.getString("userId"))
                        .videoId(jsonObject.getString("videoId"))
                        .playTotalSec(jsonObject.getLong("playSec"))
                        .playCount(1L)
                        .ts(jsonObject.getLong("ts"))
                        .build();
                out.collect(bean);
            }
        });
        //dwdPlayBeanDs.print("&&&");
        //关联求章节id，章节名称
        //TODO 6.按照uid进行分组
        KeyedStream<DwsLearnChapterPlayWindowBean, String> keyByDs = dwdPlayBeanDs.keyBy(DwsLearnChapterPlayWindowBean::getUserId);
        //keyByDs.print("****");
        //TODO 7.使用Flink的状态编程  判断是否为加购独立用户,用于后面窗口计算  [jsonObj->实体类对象]
        SingleOutputStreamOperator<DwsLearnChapterPlayWindowBean> valueStateDs = keyByDs.process(new KeyedProcessFunction<String, DwsLearnChapterPlayWindowBean, DwsLearnChapterPlayWindowBean>() {
            private ValueState<String> valueState;

            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<String> stateDescriptor =
                        new ValueStateDescriptor<String>("valueState", String.class);
                stateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());
                valueState = getRuntimeContext().getState(stateDescriptor);
            }

            @Override
            public void processElement(DwsLearnChapterPlayWindowBean value, KeyedProcessFunction<String, DwsLearnChapterPlayWindowBean, DwsLearnChapterPlayWindowBean>.Context ctx, Collector<DwsLearnChapterPlayWindowBean> out) throws Exception {
                String lastValue = valueState.value();
                String curValue = DateFormatUtil.toDate(value.getTs());
                if (lastValue == null || lastValue.compareTo(curValue) > 0) {
                    value.setPlayUuCount(1L);
                    valueState.update(curValue);
                } else {
                    value.setPlayUuCount(0L);
                }
                out.collect(value);
            }
        });
        //valueStateDs.print("bbb");
        //TODO 5.指定Watermark以及提取事件时间字段 [Flink指定水位线时要求指定为毫秒,而binlog的时间为秒]
        SingleOutputStreamOperator<DwsLearnChapterPlayWindowBean> wmDs = valueStateDs.assignTimestampsAndWatermarks(
                WatermarkStrategy.<DwsLearnChapterPlayWindowBean>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<DwsLearnChapterPlayWindowBean>() {
                            @Override
                            public long extractTimestamp(DwsLearnChapterPlayWindowBean element, long recordTimestamp) {
                                return element.getTs();
                            }
                        })
        );
        //TODO 8.分组开窗聚合 //按视频ID分组，求播放次数，总时长,人数
        SingleOutputStreamOperator<DwsLearnChapterPlayWindowBean> reduceDs = wmDs.keyBy(DwsLearnChapterPlayWindowBean::getVideoId)
                .window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10L)))
                .reduce(new ReduceFunction<DwsLearnChapterPlayWindowBean>() {
                            @Override
                            public DwsLearnChapterPlayWindowBean reduce(DwsLearnChapterPlayWindowBean value1, DwsLearnChapterPlayWindowBean value2) throws Exception {
                                value1.setPlayCount(value1.getPlayCount() + value2.getPlayUuCount());
                                value1.setPlayUuCount(value1.getPlayUuCount() + value2.getPlayUuCount());
                                value1.setPlayTotalSec(value1.getPlayTotalSec() + value2.getPlayTotalSec());
                                return value1;
                            }
                        }, new WindowFunction<DwsLearnChapterPlayWindowBean, DwsLearnChapterPlayWindowBean, String, TimeWindow>() {
                            @Override
                            public void apply(String s, TimeWindow window, Iterable<DwsLearnChapterPlayWindowBean> input, Collector<DwsLearnChapterPlayWindowBean> out) throws Exception {
                                String stt = DateFormatUtil.toYmdHms(window.getStart());
                                String edt = DateFormatUtil.toYmdHms(window.getEnd());
                                String curDate = DateFormatUtil.toDate(window.getStart());
                                for (DwsLearnChapterPlayWindowBean bean : input) {
                                    bean.setStt(stt);
                                    bean.setEdt(edt);
                                    bean.setCurDate(curDate);
                                    out.collect(bean);
                                }
                            }
                        }
                );
        //TODO 9.补充维度信息
        SingleOutputStreamOperator<DwsLearnChapterPlayWindowBean> pIdDs = AsyncDataStream.unorderedWait(
                reduceDs,
                new DimAsyncFunction<DwsLearnChapterPlayWindowBean>("dim_video_info") {
                    @Override
                    public void join(DwsLearnChapterPlayWindowBean obj, JSONObject dimInfoJsonObj) {
                        obj.setChapterId(dimInfoJsonObj.getString("chapter_id"));
                    }

                    @Override
                    public String getKey(DwsLearnChapterPlayWindowBean obj) {
                        return obj.getVideoId();
                    }
                },
                60, TimeUnit.SECONDS

        );
        SingleOutputStreamOperator<DwsLearnChapterPlayWindowBean> chapterNameDs = AsyncDataStream.unorderedWait(
                pIdDs,
                new DimAsyncFunction<DwsLearnChapterPlayWindowBean>("dim_chapter_info") {
                    @Override
                    public void join(DwsLearnChapterPlayWindowBean obj, JSONObject dimInfoJsonObj) {
                        obj.setChapterName(dimInfoJsonObj.getString("chapter_name"));
                    }

                    @Override
                    public String getKey(DwsLearnChapterPlayWindowBean obj) {
                        return obj.getChapterId();
                    }
                }
                , 60, TimeUnit.SECONDS
        );

        //TODO 10.将聚合结果写到Doris
        pIdDs.print("aaa");
        chapterNameDs.map(new MapFunction<DwsLearnChapterPlayWindowBean, String>() {
            @Override
            public String map(DwsLearnChapterPlayWindowBean value) throws Exception {
                SerializeConfig config = new SerializeConfig();
                config.propertyNamingStrategy = PropertyNamingStrategy.SnakeCase;
                String jsonString = JSON.toJSONString(value, config);
                return jsonString;
            }
        }).sinkTo(DorisUtil.getDorisSink("dws_play_chapter_view_window"));
        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
